package cn.tannn.trpc.registry.t;

import cn.tannn.tregistry.client.ServerClient;
import cn.tannn.trpc.common.api.RegistryCenter;
import cn.tannn.trpc.common.exception.ExceptionCode;
import cn.tannn.trpc.common.exception.TrpcException;
import cn.tannn.trpc.common.listener.ChangedListener;
import cn.tannn.trpc.common.listener.Event;
import cn.tannn.trpc.common.meta.InstanceMeta;
import cn.tannn.trpc.common.meta.ServiceMeta;
import cn.tannn.trpc.common.properties.rc.Connect;
import cn.tannn.trpc.common.properties.rc.RegistryCenterProperties;

import java.util.*;

import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;

import java.util.stream.Collectors;

/**
 * 自己的注册中心
 *
 * @author <a href="https://tannn.cn/">tan</a>
 * @date 2024/4/23 13:01
 * @see <a href="https://gitee.com/tanoo/tregistry">tregistry</a>
 */
@Slf4j
public class TregistryCenter implements RegistryCenter {

    private final RegistryCenterProperties rcp;

    /**
     * tregistry 注册中心客户端
     */
    private ServerClient client = null;


    public TregistryCenter(RegistryCenterProperties rcp) {
        this.rcp = rcp;
    }




    @Override
    public void start() {
        if (client != null) {
            log.info("t registry  alive ... ");
            return;
        }
        log.info("t registry Server start ...");

        Connect[] connect = rcp.getConnect();
        if (connect == null || connect.length == 0) {
            // "请填写注册中心连接信息"
            throw new TrpcException(ExceptionCode.REGISTRY_NOT_SETTING);
        } else {
            String serverList = Arrays.stream(connect).map(Connect::connectString).collect(Collectors.joining(","));
            client = new ServerClient(serverList);
        }
        client.start();
        log.info(" ===> t registry  client starting.");
    }

    @Override
    public void stop() {
        log.info(" ===> t registry  client stopped.");
        client.stop();
    }

    @Override
    public void register(ServiceMeta service, InstanceMeta instance) {
        client.reg(service.toPath(), JSON.toJSONString(instance));
        log.info(" ===> register to tr instance: {}", instance.toPath());
    }

    @Override
    public void unregister(ServiceMeta service, InstanceMeta instance) {
        client.unreg(service.toPath(), JSON.toJSONString(instance));
        log.info(" ===> unregister from tr: {}", instance.toPath());
    }

    @Override
    public List<InstanceMeta> fetchAll(ServiceMeta service) {
        List<cn.tannn.tregistry.core.model.InstanceMeta> nodes = client.findAll(service.toPath());
        if (log.isDebugEnabled()) {
            log.debug("fetchAll nodes : {}", nodes);
        }
        if (nodes == null) {
            return Collections.emptyList();
        }
        return nodes.stream().map(node -> {
            InstanceMeta meta = new InstanceMeta();
            meta.setSchema(node.getSchema());
            meta.setHost(node.getHost());
            meta.setPort(node.getPort());
            meta.setContext(node.getContext());
            meta.setParameters(node.getParameters());
            return meta;
        }).collect(Collectors.toList());
    }

    @Override
    public void subscribe(ServiceMeta service, ChangedListener changedListener) {
        log.info("rt subscribe service: {}", service.toPath());
        client.subscribe(service.toPath(), () -> {
            // 节点变动，这里会感知到
            log.info(" ====> rt  subscribe  change service: {}", service.toPath());
            List<InstanceMeta> nodes = fetchAll(service);
            changedListener.fire(new Event(nodes));
        });
    }
}
